package ru.unlocker.reports.kafka;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import java.io.IOException;
import java.io.Serial;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.val;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import ru.unlocker.reports.domain.RawReportEvent;
import ru.unlocker.reports.domain.UuidUtils;

@RequiredArgsConstructor
public class RawReportEventSchema implements KafkaRecordDeserializationSchema<RawReportEvent>,
    KafkaRecordSerializationSchema<RawReportEvent> {

  @Serial
  private static final long serialVersionUID = -5349563290624077427L;
  private static final ObjectMapper objectMapper = (new ObjectMapper())
      .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
      .registerModule(new JavaTimeModule());
  @NonNull
  private final String topicName;

  @Override
  @SneakyThrows
  public ProducerRecord<byte[], byte[]> serialize(RawReportEvent element, KafkaSinkContext context, Long timestamp) {
    return new ProducerRecord<>(
        topicName,
        UuidUtils.asBytes(element.getReportId()),
        objectMapper.writeValueAsBytes(element)
    );
  }

  @Override
  public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<RawReportEvent> out) throws IOException {
    val element = objectMapper.readValue(record.value(), RawReportEvent.class);
    out.collect(element);
  }

  @Override
  public TypeInformation<RawReportEvent> getProducedType() {
    return TypeInformation.of(RawReportEvent.class);
  }
}
